package com.ddxz.flink.state

import org.apache.flink.api.common.functions.RichFlatMapFunction
import org.apache.flink.api.common.state.{StateTtlConfig, ValueState, ValueStateDescriptor}
import org.apache.flink.api.common.time.Time
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

class CollectStateAvg extends RichFlatMapFunction[(Long, Long), (Long, Long)]{
  private var sum: ValueState[(Long, Long)] = _

  override def flatMap(input: (Long, Long), out: Collector[(Long, Long)]): Unit = {
    val tmpCurrentSum = sum.value
    val currentSum = if(null != tmpCurrentSum) {
      tmpCurrentSum
    } else {
      (0L, 0L)
    }

    val newSum = (currentSum._1 + 1, currentSum._2 + input._2)
    sum.update(newSum)

    if(newSum._1 >= 2) {
      val tmp = newSum._2 / newSum._1
      out.collect((input._1, tmp))
//      sum.clear()
    }
  }

  override def open(parameters: Configuration): Unit = {
    val descriptor = new ValueStateDescriptor[(Long, Long)] ("average", createTypeInformation[(Long, Long)])
    val ttlConfig = StateTtlConfig.newBuilder(Time.seconds(1))
      .neverReturnExpired
      .cleanupFullSnapshot
      .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
      .build
    descriptor.enableTimeToLive(ttlConfig)
    sum = getRuntimeContext.getState(descriptor)
  }
}
